package arithmetic.functionTest.myStream;

import java.util.Iterator;
import java.util.Objects;
import java.util.function.Supplier;

/**
 * 流的管道处理器(简化版)
 * 没有分隔迭代器,没有支持各种处理特性(排序)
 * Created on 2020-04-26
 */
public abstract class MyAbstractPipeline<E_IN, E_OUT> extends MyPipelinHelper<E_OUT> implements MyBaseStream<E_IN>{

    private static final String MSG_STREAM_LINKED = "stream has already been operated upon or closed";
    private static final String MSG_CONSUMED = "source already consumed or closed";

    /**
     * 源阶段
     */
    private final MyAbstractPipeline sourceStage;

    /**
     * 上一阶段
     * note:与源阶段互斥,
     */
    private final MyAbstractPipeline previousStage;

    /**
     * 下一阶段
     */
    @SuppressWarnings("rawtypes")
    private MyAbstractPipeline nextStage;

    /**
     * 深度,便于后面将所有操作串起来
     */
    private int depth;

    /**
     * 迭代器,存储集合元素
     */
    private Iterator<?> sourceIterator;

    /**
     * 是否被链接过或消费过
     */
    private boolean linkedOrConsumed;

    /**
     * 接收一个迭代器(函数式)  方便函数式提供
     */
    private Supplier<? extends Iterator<?>> sourceSupplier;

    /**
     * 源阶段构造
     */
    MyAbstractPipeline(Iterator<?> source) {
        this.previousStage = null;
        this.sourceIterator = source;
        this.sourceStage = this;
        this.depth = 0;
    }

    /**
     * 兼容下函数式入参
     */
    MyAbstractPipeline(Supplier<? extends Iterator<?>> source) {
        this.previousStage = null;
        this.sourceSupplier = source;
        this.sourceStage = this;
        this.depth = 0;
    }


    /**
     * 中间阶段构造器
     * 形成双向链表
     */
    MyAbstractPipeline(MyAbstractPipeline<?, E_IN> previousStage) {
        // 流只能用一次
        if (previousStage.linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        previousStage.linkedOrConsumed = true;
        previousStage.nextStage = this;

        this.previousStage = previousStage;
        this.sourceStage = previousStage.sourceStage;
        this.depth = previousStage.depth + 1;
    }

    /**
     * 包装sink,抽象方法,
     * 所以每个继承此类都应该实现这个方法
     */
    abstract MySink<E_IN> opWrapSink(MySink<E_IN> sink);


    /**
     * 终止操作 去调用, 例如 forEach, collect
     * @param sink
     * @param iterator
     * @param <P_IN>
     * @param <S>
     * @return
     */
    final <P_IN, S extends MySink<E_OUT>> S wrapAndCopyInto(S sink, Iterator<P_IN> iterator) {
        // wrapSink
        MySink wrappedSink = wrapSink(Objects.requireNonNull(sink));

        // copyInto
        Objects.requireNonNull(wrappedSink);
        //wrappedSink.begin(spliterator.getExactSizeIfKnown());
        iterator.forEachRemaining(wrappedSink);
        wrappedSink.end();
        return sink;
    }

    final <P_IN> MySink<P_IN> wrapSink(MySink<E_OUT> sink) {
        Objects.requireNonNull(sink);
        // 一直向前迭代,把所有操作串起来
        for (@SuppressWarnings("rawtypes") MyAbstractPipeline p = MyAbstractPipeline.this; p.depth > 0;
                p = p.previousStage) {
            sink = p.opWrapSink(sink);
        }
        return (MySink<P_IN>) sink;
    }

   /* final <P_IN> void copyInto(MySink<P_IN> wrappedSink, Iterator<P_IN> iterator) {
        Objects.requireNonNull(wrappedSink);
        //wrappedSink.begin(spliterator.getExactSizeIfKnown());
        iterator.forEachRemaining(wrappedSink);
        wrappedSink.end();
        //先判断是否短路:
       *//* if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
        }
        else {
            copyIntoWithCancel(wrappedSink, spliterator);
        }*//*
    }*/

    final <R> R evaluate(MyTerminalOp<E_OUT, R> terminalOp) {
        //assert getOutputShape() == terminalOp.inputShape();
        if (linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        linkedOrConsumed = true;
        return terminalOp.evaluateSequential( this, iterator());
    }

    /**
     * 获取迭代器
     * @return
     */
    public Iterator<E_IN> iterator() {
        // Get the source spliterator of the pipeline
        Iterator<E_IN> spliterator = null;
        if (sourceStage.sourceIterator != null) {
            spliterator = sourceStage.sourceIterator;
            sourceStage.sourceIterator = null;
        }
        else if (sourceStage.sourceSupplier != null) {
            spliterator = (Iterator<E_IN>) sourceStage.sourceSupplier.get();
            sourceStage.sourceSupplier = null;
        }
        else {
            throw new IllegalStateException(MSG_CONSUMED);
        }
        return spliterator;
    }



}
